#include "MysqlConnectionPool.h"
MysqlConnectionPool *MysqlConnectionPool::instance_ = nullptr;

MysqlConnectionPool::MysqlConnection::MysqlConnection(MYSQL *con, boost::asio::io_context &io) : con_(con), io_(io), is_run(false)
{
    t_ = std::thread([this]
                     {
                while (true)
                {
                    LOG_I<<"mysql thread run:"<<is_run;
                    std::unique_lock<std::mutex> lock(mutex_);
                    cv_.wait(lock, [this] { return is_run; });
                    is_run = false;
                    
                    int iTempRet = mysql_real_query(con_, sql_.c_str(), sql_.length());
                    if(iTempRet != 0)
                    {
                        retry_times++;
                        if(retry_times >= MAX_RETRY_TIMES)
                        {
                            LOG_E << "mysql_real_query error: " << mysql_error(con_);
                            retry_times = 0;
                            boost::asio::post(io_,[this]{
                                call_(mysql_errno(con_), nullptr);
                            });
                            continue;
                        }
                        continue;
                    }else{
                        retry_times = 0;
                        // 读取结果集
                        MYSQL_RES *res = mysql_store_result(con_);
                        if(res){
                            auto rowCount = mysql_num_rows(res);
                            auto fieldCount = mysql_num_fields(res);
                            auto result_ptr = std::make_shared<QueryResult>(res, rowCount, fieldCount);
                            MYSQL_ROW row;
                            while ((row = mysql_fetch_row(res))) {
                                    // Process each row
                                    for(int i = 0; i < fieldCount; i++) {
                                        std::cout << row[i] << " ";
                                    }
                                }
                            boost::asio::post(io_,[this,result_ptr]{
                                call_(0, result_ptr);
                            });
                        }else{
                            LOG_E << "mysql_real_query error: " << mysql_error(con_);
                            boost::asio::post(io_,[this]{
                                call_(0, nullptr);
                            });
                        }
                    }
                } });
    // t.join();
}

MysqlConnectionPool::MysqlConnectionPool(boost::asio::io_context &io, size_t size, const std::string &host, const std::string &user, const std::string &pwd, const std::string &dbname)
    : io_(io), size_(size), host_(host), user_(user), pwd_(pwd), db_(dbname)
{
    if (size_ == 0 || size_ > std::thread::hardware_concurrency())
    {
        size_ = std::thread::hardware_concurrency();
    }
    LOG_I << "Mariadb Version: " << MARIADB_CLIENT_VERSION_STR;
    LOG_I << "thread hardware_concurrency: " << size_;
    LOG_I << "mysql host: " << host_;
    LOG_I << "mysql user: " << user_;
    LOG_I << "mysql pwd: " << pwd_;
    LOG_I << "mysql db: " << db_;
    for (size_t i = 0; i < size_; ++i)
    {
        LOG_I << "mysql init: ";
        // 创建并初始化mysql对象
        MYSQL *con = mysql_init(nullptr);
        // 连接数据库
        if (!mysql_real_connect(con, host_.c_str(), user_.c_str(), pwd_.c_str(), db_.c_str(), 0, nullptr, 0))
        {
            mysql_close(con);
            std::string error_msg = "mysql_real_connect error: " + std::string(mysql_error(con));
            LOG_E << error_msg;
            throw std::runtime_error(error_msg);
        }
        // 　设置编码格式
        mysql_options(con, MYSQL_SET_CHARSET_NAME, "utf8");
        my_bool t = 1;
        mysql_options(con, MYSQL_OPT_RECONNECT, &t);
        connections_.push_back(con);
        qconnections_.push(std::make_shared<MysqlConnection>(con, io_));
    }
}

MysqlConnectionPool::~MysqlConnectionPool()
{
    // 销毁mysql对象
    for (auto &con : connections_)
    {
        mysql_close(con);
    }
}

void MysqlConnectionPool::process_queries()
{
    if (!qevents_.empty() && !qconnections_.empty())
    {
        auto event = qevents_.front();
        auto con = qconnections_.front();
        qevents_.pop();
        qconnections_.pop();
        con->query(event.first, [this, con](const error_code ec, const QueryResultPtr res)
                   {
                     boost::asio::post(io_,[this, con,ec,res]{
                        // event.second(ec, res);
                       }); 
                       qconnections_.push(con);
                       qevents_.pop();
                       process_queries(); });
    }
}
